package com.huawei.bigdata.flink.examples.utils;

import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.entity.ContentType;
import org.apache.http.nio.entity.NStringEntity;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Collections;
import java.util.Map;

public class CreateIndex extends Thread{
    private static final Logger LOG = LoggerFactory.getLogger(CreateIndex.class);
    public static void index(RestClient restClient , String mappingPath, String indexNamePrefix) throws IOException {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                while (true){
                    String indexName = null;
                    long start = System.currentTimeMillis();
                    Calendar calendar = Calendar.getInstance();
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy_MM");
                    indexName =indexNamePrefix+sdf.format(calendar.getTime());
                    Map<String, String> params = Collections.singletonMap("pretty", "true");
                    //创建指定primary shard和replica数目的索引。
                    //以下代码中的变量index、shardNum、replicaNum ，即需要创建的索引名称、索引的primary shard和replica数目。
                    //在样例工程的conf目录下的“es-example.properties”配置文件中设置，，“3”，“1”，可自行修改。
                    Response rsp = null;
                    try{
                        String path = CreateIndex.class.getClassLoader().getResource(mappingPath).getPath();
                        BufferedReader bufferedReader = new BufferedReader(new FileReader(new File(path)));
                        StringBuffer stringBuffer = new StringBuffer();
                        String line = null;
                        while ((line =bufferedReader.readLine())!=null){
                            stringBuffer.append(line+"\n\r");
                        }
                        HttpEntity entity = new NStringEntity(stringBuffer.toString(), ContentType.APPLICATION_JSON);
                        rsp = restClient.performRequest("PUT", "/" + indexName, params, entity);
                        if(HttpStatus.SC_OK == rsp.getStatusLine().getStatusCode()) {
                            System.out.println("CreateIndexWithShardNum successful.:"+indexName);
                            LOG.info("CreateIndexWithShardNum successful.:" + indexName );
                        }else {
                            LOG.error("CreateIndexWithShardNum failed.");
                        }
                    }catch (Exception e) {
                        LOG.error("CreateIndexWithShardNum failed, exception occurred.",e);
                    }
                    calendar.add(Calendar.MONTH,1);
                    calendar.set(Calendar.DAY_OF_MONTH,1);
                    calendar.set(Calendar.HOUR_OF_DAY,0);
                    calendar.set(Calendar.MINUTE,0);
                    calendar.set(Calendar.SECOND,0);
                    long sleep = calendar.getTime().getTime()-start;
                    try {
                        Thread.sleep(sleep);
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            }
        };
        Thread thread = new Thread(runnable);
        thread.start();
    }
}

